package com.springboot.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Properties;

/**
 * Kafka 消息者Demo实例
 *
 * @author: zr
 * @date: 2022-04-09
 */
public class KafkaDemoConsumer {

    public static void main(String[] args) {
        String topicName = "test-topic";
        String groupId = "test-group";

        KafkaConsumer<String, String> consumer = createKafkaConsumer();
        consumer.subscribe(Arrays.asList(topicName));
        try {
            while (true) {
                // 超时时间
                ConsumerRecords<String, String> records = consumer.poll(Integer.MAX_VALUE);
                for (ConsumerRecord<String, String> record : records) {
                    System.out.println(record.offset() + "," + record.key() + "," + record.value());
                }
            }
        } catch (Exception e) {

        }
    }

    /**
     * 创建一个Kafka消费者实例
     *
     * @param
     * @return {@link KafkaConsumer< String, String>}
     * @throws
     * @date 2022-04-09
     */
    private static KafkaConsumer<String, String> createKafkaConsumer() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "groupId");
        props.put("enable.auto.commit", true);
        props.put("auto.commit.ineterval.ms", "1000");
        // 每次重启都是从最早的offset开始读取，不是接着上一次
        props.put("auto.offset.reset", "earliest");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.SttringDeserializer");
        return new KafkaConsumer<String, String>(props);
    }
}
